package com.cl.ks.controller;


import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.ks.flow.handler.SqlQueryHandler;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.factory.SparkNodeFactory;
import com.cl.ks.entity.SparkNode;
import com.cl.ks.flow.KsFlowGraphHandler;
import com.cl.ks.service.SparkNodeService;
import com.cl.ks.utils.CommonResult;
import com.cl.mvc.controller.BaseController;
import org.apache.spark.sql.Row;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import javax.servlet.http.HttpServletResponse;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

@RestController
@RequestMapping("SparkNode")
public class SparkNodeController extends BaseController<SparkNode> {

    @Autowired
    SqlQueryHandler sqlQueryHandler;
    @Autowired
    KsFlowGraphHandler ksFlowGraphHandler;
    private static final AtomicInteger count = new AtomicInteger(0);

    private final SparkNodeService service;

    public SparkNodeController(SparkNodeService service) {
        super(service);
        this.service = service;
    }

    @GetMapping("/previewSqlQuery")
    public CommonResult<List<JSONObject>> previewSqlQuery(@RequestParam("dbCode") String dbCode, @RequestParam("sql") String sql) {
        return CommonResult.success(sqlQueryHandler.previewSqlQuery(dbCode, sql));
    }

    @PostMapping("/getFields")
    public CommonResult<JSONArray> getFields(@RequestBody SparkParam sparkParam) {
        return service.getFields(sparkParam);
    }

    @GetMapping("/test")
    public JSONObject test() {
        JSONObject res = new JSONObject();
        JSONArray items = new JSONArray();
        JSONObject item = new JSONObject();
        item.put("test", count.getAndIncrement());
        items.add(item);
        res.put("data", items);
        return res;
    }

    @GetMapping("/test1")
    public int test1() {
        return count.get();
    }

    @PostMapping("/saveCustom")
    public CommonResult<SparkNode> saveCustom(@RequestBody SparkNode sparkNode, @RequestHeader(name = "Authorization", defaultValue = "") String auth) {
        return CommonResult.success(service.saveCustom(sparkNode));
    }

    @GetMapping("/handleKsFlowGraph")
    public CommonResult handleKsFlowGraph(@RequestParam("ksFlowGraphId") Integer ksFlowGraphId,@RequestParam(value = "testFlag",defaultValue = "false") Boolean testFlag, @RequestHeader(name = "Authorization", defaultValue = "") String auth) {
        return service.handleKsFlowGraphById(ksFlowGraphId,testFlag, auth);
    }

    @GetMapping("/handleAllKsFlowGraph")
    public CommonResult handleAllKsFlowGraph(@RequestHeader(name = "Authorization", defaultValue = "") String auth) {
        return service.handleAllKsFlowGraph(auth);
    }

    @GetMapping("/filterToKsFlowGraph")
    public void filterToKsFlowGraph(@RequestParam("filterId") Integer filterId) {
        service.filterToKsFlowGraph(filterId);
    }

    @GetMapping("/allFilterToKsFlowGraph")
    public void allFilterToKsFlowGraph() {
        service.allFilterToKsFlowGraph();
    }
//
//
//    @PostMapping("/handleSparkNode")
//    public CommonResult handleSparkNode(@RequestBody SparkNode sparkNode) {
//        return CommonResult.success(service.handleSparkNode(sparkNode,null));
//    }

    @PostMapping("/downloadData")
    public CommonResult downloadData(@RequestBody List<Map<String, Object>> data, HttpServletResponse response) {
        return service.downloadData(data, response);
    }

    @GetMapping("/downloadData")
    public CommonResult downloadData(@RequestParam("ksFlowGraphId") Integer ksFlowGraphId, HttpServletResponse response) {
        return service.downloadData(ksFlowGraphId, response);
    }

    @GetMapping("/getProgress")
    public CommonResult getProgress(@RequestParam(value = "ksFlowGraphId", defaultValue = "0") Integer ksFlowGraphId) {
        return CommonResult.success(ksFlowGraphHandler.getProgress( ksFlowGraphId));
    }
}
